package com.tbit.uqbike.center.config.rabbitmq;

import cn.hutool.core.util.StrUtil;
import com.tbit.common.constant.CommonConstant;
import com.tbit.common.utils.TraceUtil;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * @author chenzhuang
 * @Description:
 * @date 2022/7/28 10:23
 */
@Configuration
public class RabbitConfig {
    @Value("${mq.exchange}")
    private String MQ_EXCHANGE;
    @Value("${mq.queue.ter_position}")
    private String MQ_QUEUE_TERPOSITION;
    @Value("${mq.queue.ter_battery}")
    private String MQ_QUEUE_TERBATTERY;
    @Value("${mq.queue.ter_lithium}")
    private String MQ_QUEUE_TERLITHIUM;


    @Bean
    public CachingConnectionFactory connectionFactory(RabbitProperties rabbitProperties) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitProperties.getHost(), rabbitProperties.getPort());
        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.setVirtualHost(StringUtils.hasText(rabbitProperties.getVirtualHost()) ? rabbitProperties.getVirtualHost() : "/");
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin amqpAdmin(CachingConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        /*rabbitAdmin.declareExchange(exchange());
        rabbitAdmin.declareQueue(mqQueueTerPosition());
        rabbitAdmin.declareQueue(mqQueueTerBattery());
        rabbitAdmin.declareQueue(mqQueueTerLithium());
        rabbitAdmin.declareBinding(bindExchange1());
        rabbitAdmin.declareBinding(bindExchange2());
        rabbitAdmin.declareBinding(bindExchange3());*/
        return rabbitAdmin;
    }

    @Bean(name = {"rabbitListenerContainerFactory"})
    @ConditionalOnProperty(
            prefix = "spring.rabbitmq.listener",
            name = {"type"},
            havingValue = "simple",
            matchIfMissing = true
    )
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, CachingConnectionFactory connectionFactory, ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory,connectionFactory);
        simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);

        // 使用RabbitTemplate接收消息前会调用这个方法
        factory.setAfterReceivePostProcessors(message -> {
            Object header = message.getMessageProperties().getHeader(CommonConstant.TRACE_ID);
            String traceId = String.valueOf(header);
            if (Objects.isNull(header) || StrUtil.isBlank(traceId)){
                traceId = TraceUtil.buildTraceId();
            }
            TraceUtil.setTraceId(traceId);
            return message;
        });
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        rabbitTemplate.setBeforePublishPostProcessors(new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                return message;
            }

            // 使用RabbitTemplate发送消息前会调用这个方法
            @Override
            public Message postProcessMessage(Message message, Correlation correlation, String exchange, String routingKey) {
                String traceId = TraceUtil.getTraceIdByLocal();
                if (StrUtil.isBlank(traceId)){
                    traceId = TraceUtil.buildTraceId();
                }
                message.getMessageProperties().setHeader(CommonConstant.TRACE_ID,traceId);
                return message;
            }
        });

        rabbitTemplate.setExchange(MQ_EXCHANGE);
        return rabbitTemplate;
    }

    @Bean
    public Queue mqQueueTerPosition() {
        return new Queue(MQ_QUEUE_TERPOSITION, false, true, true);
    }

    @Bean
    public Queue mqQueueTerBattery() {
        return new Queue(MQ_QUEUE_TERBATTERY, false, true, true);
    }

    @Bean
    public Queue mqQueueTerLithium() {
        return new Queue(MQ_QUEUE_TERLITHIUM, false, true, true);
    }


    @Bean
    public Exchange exchange() {
        /** direct交换机 不显式指定路由键则默认为队列名 */
        return new DirectExchange(MQ_EXCHANGE, false, false);
    }

    @Bean
    public Binding bindExchange1() {
        return BindingBuilder.bind(mqQueueTerPosition()).to(exchange()).with(MQ_QUEUE_TERPOSITION).noargs();
    }

    @Bean
    public Binding bindExchange2() {
        return BindingBuilder.bind(mqQueueTerBattery()).to(exchange()).with(MQ_QUEUE_TERBATTERY).noargs();
    }

    @Bean
    public Binding bindExchange3() {
        return BindingBuilder.bind(mqQueueTerLithium()).to(exchange()).with(MQ_QUEUE_TERLITHIUM).noargs();
    }


}
